package com.atguigu.day04;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author Felix
 * @date 2024/7/12
 * 该案例演示了富函数
 * 需求：从指定的网络端口读取数据，在flink程序中，给每条数据加一个"atguigu->"前缀
 * 在使用算子对流中数据进行处理的时候，算子需要接收处理函数作为参数
 *      例如：map(MapFunction)、 filter(FilterFunction)、flatMap(FlatMapFunction)
 *      默认情况下，处理函数的类型声明的是接口类型
 *      在Flink中，针对每一个处理函数，都有一个对应的类，这个类的形式 Rich + 接口名 ，例如MapFunction->RichMapFunction
 *      我们称这样的类为富函数类
 *      富函数可以通过运行时上下文获取更丰富的信息
 *      富函数是带有生命周期的,以map算子为例 map(RichMapFunction)
 *          open
 *              在生命周期开始的时候执行的方法，一般用于做资源的初始化
 *              在每个并行度(并行子任务)上执行一次
 *          map
 *              流中每来一条数据，执行一次
 *          close
 *              在生命周期结束的时候执行的方法，一般用于做资源的关闭
 *              如果处理的是无界数据，close方法将永远不会被执行
 *              在每个并行度(并行子任务)上执行一次
 */
public class Flink01_Rich_Function {
    public static void main(String[] args) throws Exception {
        //TODO 1.指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.从指定的网络端口读取数据
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        //TODO 3.对读取的数据进行处理  加前缀
        /*SingleOutputStreamOperator<String> mapDS = socketDS.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                return "atguigu->" + value;
            }
        });*/

        SingleOutputStreamOperator<String> mapDS = socketDS.map(new RichMapFunction<String, String>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("~~~open生命周期开始~~~");
                RuntimeContext runtimeContext = getRuntimeContext();
                int index = runtimeContext.getIndexOfThisSubtask();
                String name = runtimeContext.getTaskNameWithSubtasks();
                System.out.println("索引" + index + "为的名字是" + name + "子任务执行了");
            }

            @Override
            public void close() throws Exception {
                System.out.println("~~~close生命周期结束~~~");
            }

            @Override
            public String map(String value) throws Exception {
                return "atguigu->" + value;
            }
        });

        //TODO 4.打印输出
        mapDS.print();
        //TODO 5.提交作业
        env.execute();
    }
}
